-----查看所有操作
---转运中心卸车
with t_print_reback as (
    select
        waybill_no ,print_network_code
    from
        jms_dwd.dwd_tab_reback_transfer_express_base
    where print_time between date_add('{{ execution_date | date_add(-1) | cst_ds }}',-5) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',1)
      and status=2
      and print_flag=1
    group by waybill_no ,print_network_code
)
,t_unloading as(
    select  /*+ BROADCAST(rb) */
        tmp.billcode
         ,tmp.scantime
         ,tmp.inputsite
         ,tmp.scansitecode
         ,tmp.scanuser
         ,tmp.scanusercode
         ,'转运中心卸车' as scan_type   --上一站
         ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
         ,null as sn_code
         ,tmp.next_scansite_code
         ,tmp.next_scansite_name
         ,tmp.shipment_no
         ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
         ,package_code --包号
         ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
    from(
            select
                waybill_no as billcode
                 ,scan_time as scantime
                 ,scan_site as inputsite
                 ,scan_site_code as  scansitecode
                 ,scan_user as scanuser
                 ,scan_user_code as scanusercode
                 ,package_code
                 ,pre_station as next_scansite_name
                 ,pre_station_code as next_scansite_code
                 ,transfer_code as shipment_no
                 --,row_number() over(partition by t1.scansitecode,t1.scanusercode,t1.billcode order by t1.scantime) bill_rank
                 ,row_number() over(partition by scan_site_code,substr(scan_time,1,10),waybill_no order by scan_time) bill_rank
            from jms_dwd.dwd_tab_barscan_centerarrival_base_dt t1
            where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',0) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2) and length(waybill_no) <= 15
              and network_type = 2
            --and package_code is null
        ) tmp  left join t_print_reback rb on tmp.billcode=rb.waybill_no and rb.print_network_code=tmp.scansitecode
    where (bill_rank = 1 and rb.waybill_no is null) or  rb.waybill_no is not null
),
---**装车数据去包内件
     t_loading as(
         ---装车扫描
         select /*+ BROADCAST(rb) */
             tmp.billcode
              ,tmp.scantime
              ,tmp.inputsite
              ,tmp.scansitecode
              ,tmp.scanuser
              ,tmp.scanusercode
              ,'转运中心装车' as scan_type    --下一站
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
              ,null as sn_code
              ,tmp.next_scansite_code
              ,tmp.next_scansite_name
              ,tmp.shipment_no
              ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
              ,package_code --包号
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
         from(
                 select
                     waybill_no as billcode
                      ,scan_time as scantime
                      ,scan_site as inputsite
                      ,scan_site_code as  scansitecode
                      ,scan_user as scanuser
                      ,scan_user_code as scanusercode
                      ,package_code
                      ,next_station as next_scansite_name
                      ,next_station_code as next_scansite_code
                      ,transfer_code as shipment_no
                      --,row_number() over(partition by t1.scansitecode,t1.scanusercode,t1.billcode order by t1.scantime) bill_rank
                      ,row_number() over(partition by scan_site_code,substr(scan_time,1,10),waybill_no order by scan_time) bill_rank
                 from jms_dwd.dwd_tab_barscan_centersend_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',0) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2)
                   and length(waybill_no) <= 15
                   and network_type = 2
                 --and package_code is null
             ) tmp left join t_print_reback rb on tmp.billcode=rb.waybill_no and rb.print_network_code=tmp.scansitecode
         where (bill_rank = 1 and rb.waybill_no is null) or  rb.waybill_no is not null
     ),
---**建包扫描
     t_bagging as(
         select /*+ BROADCAST(rb) */
             tmp.billcode
              ,tmp.scantime
              ,tmp.inputsite
              ,tmp.scansitecode
              ,tmp.scanuser
              ,tmp.scanusercode
              ,'转运中心建包' as scan_type
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
              ,null as sn_code
              ,tmp.next_scansite_code
              ,tmp.next_scansite_name
              ,tmp.shipment_no
              ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
              ,package_code
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
         from(
                 select
                     waybill_no as billcode
                      ,scan_time as scantime
                      ,scan_site as inputsite
                      ,scan_site_code as  scansitecode
                      ,scan_user as scanuser
                      ,scan_user_code as scanusercode
                      ,package_code
                      ,next_station_code as next_scansite_code
                      ,next_station as next_scansite_name
                      ,transfer_code as shipment_no
                      ,row_number() over(partition by scan_site_code,scan_user_code,waybill_no order by scan_time) bill_rank
                 from jms_dwd.dwd_tab_barscan_bagging_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',0)
                     and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2)
                   and length(waybill_no) <= 15
                   --and inputsite like '%转运中心%'
                   and network_type = 2
                   and waybill_no not like 'B%'
             ) tmp  left join t_print_reback rb on tmp.billcode=rb.waybill_no and rb.print_network_code=tmp.scansitecode
         where (bill_rank = 1 and rb.waybill_no is null) or  rb.waybill_no is not null

     ),
---**转运中心拆包扫描
     t_loading2 as(
         ---装车扫描
         select
             billcode
              ,scantime
              ,inputsite
              ,scansitecode
              ,scanuser
              ,scanusercode
              ,'转运中心装车' as scan_type
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
              ,null as sn_code
              ,next_scansite_code
              ,next_scansite_name
              ,shipment_no
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
         from(
                 select
                     waybill_no as billcode
                      ,scan_time as scantime
                      ,scan_site as inputsite
                      ,scan_site_code as  scansitecode
                      ,scan_user as scanuser
                      ,scan_user_code as scanusercode
                      ,next_station_code as next_scansite_code
                      ,next_station as next_scansite_name
                      ,transfer_code as shipment_no
                      ,row_number() over(partition by scan_site_code,waybill_no order by scan_time,scan_user_code) bill_rank
                 from jms_dwd.dwd_tab_barscan_centersend_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',0)
                     and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2) and length(waybill_no) <= 15
                   --and inputsite like '%转运中心%'
                   and network_type = 2
             ) tmp where bill_rank = 1
     ),
     --包号取对应的拆包网点名称和编码<--电子包牌表（增量同步）
     tab_package_center as (
         select center_code,package_number
         from  jms_dwd.dwd_tab_electronic_package_list_base_dt  --每天新增
         where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}' ,-30)
                   and date_add('{{ execution_date | date_add(-1) | cst_ds }}' ,0)
         group by center_code,package_number
     ),
     --包号和单号关系<--建包扫描表
     tab_package_waybill as(
         select
             waybill_no,
             package_code,
             scan_user_code
                          scan_user,
             scan_site,
             scan_site_code,
             scan_time,
             ROW_NUMBER() over(partition by waybill_no,package_code order by scan_time desc) rk
         from  jms_dwd.dwd_tab_barscan_bagging_base_dt
         where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}' ,-7)
             and date_add('{{ execution_date | date_add(-1) | cst_ds }}' ,2)
           and package_code is not null
     ),

     ---单号和拆包机构
     tab_waybill_cernter as (
         select  tab_package_waybill.waybill_no,tab_package_waybill.package_code,tab_package_center.center_code --单号,包号，中心
         from tab_package_waybill
                  inner join  tab_package_center on tab_package_waybill.package_code=tab_package_center.package_number
     ),


     tab_unpackage as (  --发件的中心、包内件(单号)/散件到建包数据里边根据拆包末端机构对比，如果一致的情况，比对包内件(单号)/散件，如果建包数据存在相同的单号，则认定做了拆包扫描，作为拆包扫描的记录
         --发件 包、扫描地点、单号要与包牌表 包、拆包中心、建包表单号 关联取交集
         select /*+ BROADCAST(rb) */
             t.billcode
              ,t.scantime
              ,t.inputsite
              ,t.scansitecode
              ,t.scanuser
              ,t.scanusercode
              ,t.scan_type
              ,t.update_time
              ,t.sn_code
              ,t.next_scansite_code
              ,t.next_scansite_name
              ,t.shipment_no
              ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
              ,package_code
              ,t.dt
         from
             (
                 select
                     billcode
                      ,scantime
                      ,inputsite
                      ,scansitecode
                      ,scanuser
                      ,scanusercode
                      ,'转运中心拆包' as scan_type
                      ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
                      ,null as sn_code
                      ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
                      ,ROW_NUMBER() over(partition by billcode,scansitecode order by scantime,scanusercode) rk
                ,next_scansite_code
                      ,next_scansite_name
                      ,shipment_no
                      ,tab_waybill_cernter.package_code
                 from t_loading2 t_loading
                          inner join tab_waybill_cernter
                                     on t_loading.scansitecode=tab_waybill_cernter.center_code
                                         and t_loading.billcode=tab_waybill_cernter.waybill_no
             )t left join t_print_reback rb on t.billcode=rb.waybill_no and rb.print_network_code=t.scansitecode
         where (t.rk = 1 and rb.waybill_no is null) or  rb.waybill_no is not null
     ),
---**供给台扫描
     t_small_upper as(
         select /*+ BROADCAST(rb) */
             tmp.waybill_no as billcode
              ,tmp.scan_time as scantime
              ,tmp.inputsite
              ,tmp.scansitecode
              ,tmp.scanuser
              ,tmp.scanusercode
              ,'供件台扫描' as scan_type
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
              ,tmp.supply_desk_code as sn_code
              ,null as next_scansite_code
              ,null as next_scansite_name
              ,null as shipment_no
              ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
              ,null as package_code
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
         from(
                 select
                     a.waybill_no
                      ,a.scan_time
                      ,a.network_name as inputsite
                      ,a.network_code as scansitecode
                      ,null as scanuser
                      ,a.user_num as scanusercode
                      ,a.supply_desk_code --设备编码
                      --,row_number() over(partition by network_code,user_num,supply_desk_code,waybill_no order by scan_time) bill_rank
                      ,row_number() over(partition by a.dt,a.network_code,a.waybill_no order by a.scan_time desc,supply_desk_code) bill_rank
                 from jms_dwd.dwd_ass_scan_small_upper_base_dt a
                          inner join jms_dim.dim_network_whole_massage n on n.code=a.network_code
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',0) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2)
                   --and length(waybill_no) <= 15
                   and a.owned_venue_type = '335'
                   --and network_name like '%转运中心%'
                   and n.network_type=4
             ) tmp left join t_print_reback rb on tmp.waybill_no=rb.waybill_no and rb.print_network_code=tmp.scansitecode
         where (bill_rank = 1 and rb.waybill_no is null) or  rb.waybill_no is not null
     ),
---**小件交件
     t_small_collect_scan as (
         SELECT /*+ BROADCAST(rb) */
             t1.waybill_no as billcode
              ,t1.scan_time as scantime
              ,t1.scan_site as inputsite
              ,t1.scan_site_code as scansitecode
              ,t1.scan_user as scanuser
              ,t1.scan_user_code as scanusercode
              ,'小件交件扫描' as scan_type
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
              ,null as sn_code
              ,null as next_scansite_code
              ,null as next_scansite_name
              ,null as shipment_no
              ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
              ,package_code
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
         from
             (--建包扫描(中心 单号 去重)-本中心到件包内件明细, 关联条件 中心和单号
                 select
                     waybill_no ,
                     scan_site_code ,
                     scan_site ,
                     scan_time ,
                     package_code,
                     scan_user_code,
                     scan_user,
                     ROW_NUMBER() over(partition by waybill_no,scan_site_code order by scan_time,scan_user_code) rk
                 from
                     jms_dwd.dwd_tab_barscan_bagging_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',-1) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',1)
                   and network_type =2 --中心
                   and length(waybill_no) <= 15
             )t1 left join
             (
                 select
                     waybill_no
                      ,scan_site_code
                      ,row_number() over(partition by scan_site_code,waybill_no order by scan_time) bill_rank
                 from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',-1) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',1) and length(waybill_no) <= 15
                   and package_code is not null
                   and network_type = 2
             ) t2 on  t2.bill_rank = 1 and t1.waybill_no=t2.waybill_no and t1.scan_site_code=t2.scan_site_code
                 left join t_print_reback rb on t1.waybill_no=rb.waybill_no and rb.print_network_code=t1.scan_site_code
         where ((t1.rk=1 and rb.waybill_no is null) or  rb.waybill_no is not null) and t2.waybill_no is null
     ),
---**人工滑槽分拣
     t_manual_stool_scan as (
         SELECT  /*+ BROADCAST(rb) */
             t1.waybill_no as billcode
              ,t1.scan_time as scantime
              ,t1.scan_site as inputsite
              ,t1.scan_site_code as scansitecode
              ,t1.scan_user as scanuser
              ,t1.scan_user_code as scanusercode
              ,'人工滑槽分拣' as scan_type
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as update_time
              ,t1.sn_code
              ,t1.next_scansite_code
              ,t1.next_scansite_name
              ,t1.shipment_no
              ,if(rb.waybill_no is not null,1,0) as is_reback --是否是退转件
              ,t1.package_code
              ,'{{ execution_date | date_add(-1) | cst_ds }}' as dt
         from
             (select * from(  --卸车到件包内件-卸车包内件和中心建包扫描重复数据-中心到件包号和发件一致的包号
                               select
                                   waybill_no ,
                                   scan_site_code ,
                                   scan_site ,
                                   scan_time ,
                                   package_code,
                                   scan_user_code,
                                   scan_user,
                                   null as sn_code,
                                   pre_station as next_scansite_name,
                                   pre_station_code as next_scansite_code,
                                   shipment_no,
                                   ROW_NUMBER() over(partition by waybill_no,scan_site_code order by scan_time,scan_user_code) rk
                               from
                                   jms_dwd.dwd_tab_barscan_unloading_base_dt
                               where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',-1) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',1)
                                 and network_type =2 --中心
                                 and length(waybill_no) <= 15
                                 and package_code is not null
                           )t where rk=1
             )t1 left join
             ( --中心建包
                 select
                     waybill_no
                      ,scan_site_code
                      ,row_number() over(partition by scan_site_code,waybill_no order by scan_time) bill_rank
                 from jms_dwd.dwd_tab_barscan_bagging_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',-1) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2) and length(waybill_no) <= 15
                   and package_code is not null
                   and network_type = 2
             ) t2 on  t2.bill_rank = 1 and t1.waybill_no=t2.waybill_no and t1.scan_site_code=t2.scan_site_code
                 left join
             (   --中心发件单号和包号
                 select
                     waybill_no
                      ,scan_site_code
                      ,package_code
                      ,row_number() over(partition by scan_site_code,waybill_no order by scan_time) bill_rank
                 from jms_dwd.dwd_tab_barscan_centersend_base_dt
                 where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}',-1) and date_add('{{ execution_date | date_add(-1) | cst_ds }}',2) and length(waybill_no) <= 15
                   and package_code is not null
                   and network_type = 2
             ) t3 on  t3.bill_rank = 1 and t1.waybill_no=t3.waybill_no and t1.scan_site_code=t3.scan_site_code and t1.package_code=t3.package_code
                 left join t_print_reback rb on t1.waybill_no=rb.waybill_no and rb.print_network_code=t1.scan_site_code
         where ((t1.rk=1 and rb.waybill_no is null) or  rb.waybill_no is not null)
           and t2.waybill_no is null and t3.waybill_no is null
     ),

     t_union_four as (  --装车扫描
         select * from t_loading
         union all --卸车扫描
         select * from t_unloading
         union all --建包扫描
         select * from t_bagging
         union all --拆包扫描
         select * from tab_unpackage
         union all --自动分拣
         select * from t_small_upper
         union all --小件交件
         select * from t_small_collect_scan
         union all --人工滑槽
         select * from t_manual_stool_scan
     )
insert overwrite table bidefault.dm_employee_operate_num_all_detail partition(dt)
select
    billcode,scantime
     ,inputsite,scansitecode
     ,scanuser ,scanusercode
     ,scan_type,update_time,sn_code
     ,next_scansite_code,next_scansite_name,shipment_no,is_reback,package_code
     ,dt
from t_union_four
group by billcode,scantime,inputsite,scansitecode,scanuser,scanusercode,scan_type,update_time,sn_code,next_scansite_code,next_scansite_name,shipment_no,is_reback,package_code,dt
distribute by dt,abs(hash(billcode))%300;
------------------------------------------------------------------------------------
---查看员工信息+设备编码字段添加
with t_ep_ins_detail as (
    select *
    from jms_dwd.dwd_ep_create_institution_detail_base
    where date(sign_in_time) >= date_add('{{ execution_date | date_add(-1) | cst_ds }}', -5)
    ),
    t_ep_ins as (
select *
from jms_dwd.dwd_ep_create_institution_base
where date(create_day)= '{{ execution_date | date_add(-1) | cst_ds }}'
    ),
    t_ep_classes as (
select
    site_code,
    site_name,
    code as schedule_code,
    name as schedule_name,
    CONCAT('{{ execution_date | date_add(-1) | cst_ds }}',' ',start_time,':00') as schedule_start_time,
    CONCAT(date_add('{{ execution_date | date_add(-1) | cst_ds }}',cast(is_next_day as int)),' ',end_time,':00')  as schedule_end_time,
    is_next_day,
    create_time,
    status
from jms_dim.dim_ide_ep_center_classes_base
    )
--存在数据重复 一个员工签入2次
insert overwrite table bidefault.dm_employee_operate_info_detail partition(dt)
select t_ep_ins.center_code
        , t_ep_ins.center_name
        , t_ep_ins.institution_code
        , t_ep_ins.institution_name
        , t_ep_ins.institutiona_code
        , t_ep_ins.institutiona_name
        , t_ep_ins.schedule_code
        , t_ep_ins.schedule_name
        , t_ep_ins.create_day
        , t_ep_ins_detail.employ_code
        , t_ep_ins_detail.employ_name
        , t_ep_ins_detail.post_code
        , t_ep_ins_detail.post_name
        , case when t_ep_ins.institutiona_name='人工滑槽分拣' and t_ep_ins.is_special=1 then '人工滑槽分拣'
    when t_ep_ins.institutiona_name='小件交件' and t_ep_ins.is_special=1 then '小件交件'
    else t_ep_ins_detail.linked_method
    end as linked_method
        , t_ep_ins_detail.sign_in_time
        , t_ep_ins_detail.sign_out_time
        , t_ep_ins_detail.work_hours
        ,
    concat_ws(
    "#",
    nvl(
    case when t_ep_ins_detail.unpack_flag=1 then t_ep_ins.work_start_time
    when t_ep_ins.is_special=1 then t.schedule_start_time
    else t_ep_ins_detail.sign_in_time end
        , '9999-12-31 00:00:00'),  --班次或者班组或者个人开始时间
    nvl(
    case when t_ep_ins_detail.unpack_flag=1 then t_ep_ins.work_end_time
    when t_ep_ins.is_special=1 then t.schedule_end_time
    else t_ep_ins_detail.sign_out_time end
        , '1970-01-01 00:00:00'), --班次或者班组或者个人结束时间
    case when t_ep_ins.institutiona_name='人工滑槽分拣' and t_ep_ins.is_special=1 then '人工滑槽分拣'
    when t_ep_ins.institutiona_name='小件交件' and t_ep_ins.is_special=1 then '小件交件'
    else t_ep_ins_detail.linked_method end  ---都加上linked_method
    ) group_time_type
        , to_date(t_ep_ins.create_day) as update_time
        , t_ep_ins_detail.sn_code                      --设备编码添加
        , t_ep_ins.work_start_time
        , t_ep_ins.work_end_time
        , t_ep_ins_detail.unpack_flag  as unpackflag_e --t_ep_ins_detail.unpackFlag
        , t_ep_ins_detail.create_institution_id
        , t_ep_ins_detail.calculate_type --结算类型添加
        , t.schedule_start_time  --班次开始时间
        , t.schedule_end_time --班次结束时间
        , t_ep_ins.is_special --是否特殊班组：目前目前小件交件、人工滑槽属于特殊班组
        , t_ep_ins.rely_institution_code  --挂靠班组
        , t_ep_ins.rely_institution_name  --挂靠班组名称
        , t_ep_ins_detail.employ_type  --员工属性1临时工/2正式工/3小时工(已废弃)/4职能/5环节外包
        , t_ep_ins_detail.operate_code --操作属性编码 1-操作/2-非操作
        , to_date(t_ep_ins.create_day) as dt
from t_ep_ins_detail
    inner join t_ep_ins
on t_ep_ins_detail.create_institution_id = t_ep_ins.id
    left join t_ep_classes t on t_ep_ins.schedule_code=t.schedule_code  --关联班次表
distribute by dt;
------------------------------------------------------------------------------------------------
---查看员工具体操作单号+单号要加上重量字段
--计重挂靠计件 2022-10-09
-- 1、员工操作明细 挂靠班组操作明细
-- 1、计重班组+挂靠班组不为空
--    通过挂靠中心+班组获取明细 排序得到单号唯一id
--    赋值给计重员工编码 ，关联条件 挂靠班组+中心+单号编码=员工组内顺序编码+中心+挂靠班组

--set hive.mapred.mode ='proceed'
--操作扫描计件系统：转运中心卸车、转运中心装车、转运中心建包、供件台扫描
--基础表：员工表、操作表
with t_join_ep_ins as (
    select * from bidefault.dm_employee_operate_info_detail where dt = '{{ execution_date | date_add(-1) | cst_ds }}'
),
     t_all_oper as (
         select * from bidefault.dm_employee_operate_num_all_detail where dt = '{{ execution_date | date_add(-1) | cst_ds }}' ),
     t_other_scan as ( --**转运中心卸车、转运中心装车操作单号和员工关连（通过员工编码、扫描类型、扫描时间在员工签入签出时间范围内）
         select *
         from (
                  select t_all_oper.billcode
                       , t_all_oper.scan_time
                       , t_all_oper.scan_type
                       , t_join_ep_ins.center_code
                       , t_join_ep_ins.center_name
                       , t_join_ep_ins.institution_code
                       , t_join_ep_ins.institution_name
                       , t_join_ep_ins.institutiona_code
                       , t_join_ep_ins.institutiona_name
                       , t_join_ep_ins.schedule_code
                       , t_join_ep_ins.schedule_name
                       , t_join_ep_ins.create_day
                       , t_join_ep_ins.employ_code
                       , t_join_ep_ins.employ_name
                       , t_join_ep_ins.post_code
                       , t_join_ep_ins.post_name
                       , t_join_ep_ins.linked_method
                       , t_join_ep_ins.sign_in_time
                       , t_join_ep_ins.sign_out_time
                       , t_join_ep_ins.work_hours
                       , t_join_ep_ins.group_time_type
                       , to_date(t_join_ep_ins.create_day) as update_time ---修改为建组日期作为分区日期
                       , t_join_ep_ins.sn_code
                       , t_join_ep_ins.calculate_type
                       , t_all_oper.next_scansite_code
                       , t_all_oper.next_scansite_name
                       , t_all_oper.shipment_no
                       , t_all_oper.is_reback
                       , t_all_oper.package_code
                       , to_date(t_join_ep_ins.create_day) as dt
                  from t_join_ep_ins
                           left join
                       (select * from t_all_oper where scan_type!='供件台扫描' and scan_type!='转运中心拆包') t_all_oper
                       on t_join_ep_ins.center_code = t_all_oper.scansite_code
                           and t_join_ep_ins.employ_code = t_all_oper.scanuser_code
                           and t_join_ep_ins.linked_method = t_all_oper.scan_type ---按照操作方式关联
              ) tmp1
         where scan_time between sign_in_time and sign_out_time
     ),
     t_ass_scan as ( --**供给台扫描与员工关联(通过员工编码、扫描类型、设备编码、扫描时间在员工签入签出时间范围内)
         select *
         from (
                  select t_all_oper.billcode
                       , t_all_oper.scan_time
                       , t_all_oper.scan_type
                       , t_join_ep_ins.center_code
                       , t_join_ep_ins.center_name
                       , t_join_ep_ins.institution_code
                       , t_join_ep_ins.institution_name
                       , t_join_ep_ins.institutiona_code
                       , t_join_ep_ins.institutiona_name
                       , t_join_ep_ins.schedule_code
                       , t_join_ep_ins.schedule_name
                       , t_join_ep_ins.create_day
                       , t_join_ep_ins.employ_code
                       , t_join_ep_ins.employ_name
                       , t_join_ep_ins.post_code
                       , t_join_ep_ins.post_name
                       , t_join_ep_ins.linked_method
                       , t_join_ep_ins.sign_in_time
                       , t_join_ep_ins.sign_out_time
                       , t_join_ep_ins.work_hours
                       , t_join_ep_ins.group_time_type
                       , to_date(t_join_ep_ins.create_day) as update_time ---修改为建组日期作为分区日期
                       , t_join_ep_ins.sn_code
                       , t_join_ep_ins.calculate_type
                       , t_all_oper.next_scansite_code
                       , t_all_oper.next_scansite_name
                       , t_all_oper.shipment_no
                       , t_all_oper.is_reback
                       , t_all_oper.package_code
                       , to_date(t_join_ep_ins.create_day) as dt
                  from t_join_ep_ins
                           left join (select * from t_all_oper where scan_type = '供件台扫描') t_all_oper
                                     on t_join_ep_ins.center_code = t_all_oper.scansite_code --先按照员工编码关联，然后在按照设备编码关联
                                         and (t_join_ep_ins.employ_code = t_all_oper.scanuser_code or
                                              t_join_ep_ins.sn_code = t_all_oper.sn_code)
                                         ---按照操作方式关联
                                         and t_join_ep_ins.linked_method = t_all_oper.scan_type
              ) tmp1
         where scan_time between sign_in_time and sign_out_time
     ),
--**拆包扫描统计
     t_ep_ins1 as ( --拆包标识员工所在班组的信息
         select create_institution_id
              , center_code                                      --中心
              , max(center_name)               center_name
              , institution_code                                 --组织编码
              , max(institution_name)          institution_name
              , institutiona_code                                --组织编码
              , max(institutiona_name)         institutiona_name --组织编码--组织编码
              , schedule_code                                    --班次编码
              , max(schedule_name)             schedule_name
              , substr(work_start_time, 1, 19) work_start_time   --班组开始时间
              , substr(work_end_time, 1, 19)   work_end_time     --班组结束时间
         from t_join_ep_ins
         where unpackflag_e = 1
         group by create_institution_id
                , center_code       --中心
                , institution_code  --组织编码
                , institutiona_code --组织编码
                , schedule_code     --班次编码
                , work_start_time   --班组开始时间
                , work_end_time --班组结束时间
     ),
     t_unpackage_scan_schedule as ( -----拆包单号和班组信息匹配
         SELECT *,
                row_number() over( partition by create_institution_id order by scan_time) rk
         FROM (
                  select t_all_oper.billcode
                       , t_all_oper.scan_time
                       , t_all_oper.scan_type
                       , t_ep_ins1.center_code
                       , t_ep_ins1.center_name
                       , t_ep_ins1.institution_code
                       , t_ep_ins1.institution_name
                       , t_ep_ins1.institutiona_code
                       , t_ep_ins1.institutiona_name
                       , t_ep_ins1.schedule_code
                       , t_ep_ins1.schedule_name
                       , t_ep_ins1.work_start_time --班组开始时间
                       , t_ep_ins1.work_end_time   --班组结束时间
                       --,to_date(t_join_ep_ins.create_day) as update_time ---修改为建组日期作为分区日期
                       --,to_date(t_join_ep_ins.create_day) as dt
                       , t_ep_ins1.create_institution_id
                       , t_all_oper.next_scansite_code
                       , t_all_oper.next_scansite_name
                       , t_all_oper.shipment_no
                       , t_all_oper.is_reback
                       , t_all_oper.package_code
                       , row_number() over(partition by t_all_oper.billcode ,t_ep_ins1.center_code order by t_ep_ins1.work_start_time ,t_ep_ins1.institution_code ) rk1 --班组内单号排序
                  from t_ep_ins1
                           inner join (select * from t_all_oper where scan_type = '转运中心拆包') t_all_oper
                                      on t_ep_ins1.center_code = t_all_oper.scansite_code --不按员工关联，随机平均分配单号
                                          and t_all_oper.scan_time >= t_ep_ins1.work_start_time
                                          and t_all_oper.scan_time <= t_ep_ins1.work_end_time
              ) t
         where t.rk1 = 1
     ),
     t_unpackage_scan_employ as (
         select t_unpackage_scan_schedule.billcode
              , t_unpackage_scan_schedule.scan_time
              , t_unpackage_scan_schedule.scan_type
              , t_join_ep_ins.center_code
              , t_join_ep_ins.center_name
              , t_join_ep_ins.institution_code
              , t_join_ep_ins.institution_name
              , t_join_ep_ins.institutiona_code
              , t_join_ep_ins.institutiona_name
              , t_join_ep_ins.schedule_code
              , t_join_ep_ins.schedule_name
              , t_join_ep_ins.create_day
              , t_join_ep_ins.employ_code
              , t_join_ep_ins.employ_name
              , t_join_ep_ins.post_code
              , t_join_ep_ins.post_name
              , t_join_ep_ins.linked_method
              , t_join_ep_ins.sign_in_time
              , t_join_ep_ins.sign_out_time
              , t_join_ep_ins.work_hours
              , t_join_ep_ins.group_time_type
              , to_date(t_join_ep_ins.create_day) as update_time ---修改为建组日期作为分区日期
              , t_join_ep_ins.sn_code
              , t_join_ep_ins.calculate_type
              , t_unpackage_scan_schedule.next_scansite_code
              , t_unpackage_scan_schedule.next_scansite_name
              , t_unpackage_scan_schedule.shipment_no
              , t_unpackage_scan_schedule.is_reback
              , t_unpackage_scan_schedule.package_code
              , to_date(t_join_ep_ins.create_day) as dt
         from t_unpackage_scan_schedule
                  left join
              (
                  select *,
                         row_number() over(partition by create_institution_id order by sign_in_time) rk1, --给员工添加编码
                          count(1)     over(partition by create_institution_id) cnt1  --班组中有有拆包标识的员工个数
                  from t_join_ep_ins
                  where unpackflag_e = 1
              ) t_join_ep_ins --拆包员工的信息
              on t_unpackage_scan_schedule.create_institution_id = t_join_ep_ins.create_institution_id --班组对应
                  and t_unpackage_scan_schedule.rk%t_join_ep_ins.cnt1=t_join_ep_ins.rk1-1 --单号序号、员工个数取模=员工序号
    ),
     ---**小件交件和人工滑槽=标准组织部
     ---一个标准组织只有一个特殊班组一个班组会有多个班次
     ---1、求特殊标准组织的班次开始时间和结束时间
    t_ep_ins_spe as (
select create_institution_id
        , center_code                                      --中心
        , max(center_name)               center_name
        , institution_code                                 --组织编码
        , max(institution_name)          institution_name
        , institutiona_code                                --组织编码
        , max(institutiona_name)         institutiona_name --组织编码--组织编码
        , schedule_code                                    --班次编码
        , max(schedule_name)             schedule_name
        , max(work_start_time) work_start_time   --班组开始时间
        , max(work_end_time)   work_end_time     --班组结束时间
        , substr(schedule_start_time,1,19)  schedule_start_time
        , substr(schedule_end_time,1,19)  schedule_end_time
from t_join_ep_ins
where is_special = 1
group by create_institution_id
        , center_code       --中心
        , institution_code  --组织编码
        , institutiona_code --标准组织编码
        , schedule_code     --班次编码
        , schedule_start_time --班次开始时间
        , schedule_end_time --班次结束时间
    ),
    --2、用班次的时间筛选出单号
    t_special_bill as ( -----拆包单号和班组信息匹配
SELECT *,
    row_number() over( partition by create_institution_id order by scan_time) rk
FROM (
    select t_all_oper.billcode
        , t_all_oper.scan_time
        , t_all_oper.scan_type
        , t_ep_ins_spe.center_code
        , t_ep_ins_spe.center_name
        , t_ep_ins_spe.institution_code
        , t_ep_ins_spe.institution_name
        , t_ep_ins_spe.institutiona_code
        , t_ep_ins_spe.institutiona_name
        , t_ep_ins_spe.schedule_code
        , t_ep_ins_spe.schedule_name
        , t_ep_ins_spe.work_start_time --班组开始时间
        , t_ep_ins_spe.work_end_time   --班组结束时间
        , t_ep_ins_spe.schedule_start_time --班次开始时间
        , t_ep_ins_spe.schedule_end_time   --班次结束时间
        , t_ep_ins_spe.create_institution_id
        , t_all_oper.next_scansite_code
        , t_all_oper.next_scansite_name
        , t_all_oper.shipment_no
        , t_all_oper.is_reback
        , t_all_oper.package_code
    --目前认为这里不需要去重
    ---, row_number() over(partition by t_all_oper.billcode ,t_ep_ins_spe.center_code,t_ep_ins_spe.institutiona_code order by t_ep_ins_spe.schedule_start_time ,t_ep_ins_spe.institution_code,t_ep_ins_spe.schedule_code) rk1 --班组、班次、班次开始内单号排序
    from t_ep_ins_spe
    inner join (select * from t_all_oper where scan_type in ('小件交件','人工滑槽分拣')) t_all_oper
    on t_ep_ins_spe.center_code = t_all_oper.scansite_code --不按员工关联，随机平均分配单号
    and t_all_oper.scan_time >= t_ep_ins_spe.schedule_start_time
    and t_all_oper.scan_time <= t_ep_ins_spe.schedule_end_time
    and t_all_oper.scan_type = t_ep_ins_spe.institutiona_name   --单号自定操作类型=标准组织名称
    ) t
    --where t.rk1 = 1
    ),
    --3、将单号均分给对应班组中的人 (和拆包一样)
    t_special_bill_employ as (
select t_special_bill.billcode
        , t_special_bill.scan_time
        , t_special_bill.scan_type
        , t_join_ep_ins.center_code
        , t_join_ep_ins.center_name
        , t_join_ep_ins.institution_code
        , t_join_ep_ins.institution_name
        , t_join_ep_ins.institutiona_code
        , t_join_ep_ins.institutiona_name
        , t_join_ep_ins.schedule_code
        , t_join_ep_ins.schedule_name
        , t_join_ep_ins.create_day
        , t_join_ep_ins.employ_code
        , t_join_ep_ins.employ_name
        , t_join_ep_ins.post_code
        , t_join_ep_ins.post_name
        , t_join_ep_ins.linked_method
        , t_join_ep_ins.sign_in_time
        , t_join_ep_ins.sign_out_time
        , t_join_ep_ins.work_hours
        , t_join_ep_ins.group_time_type
        , to_date(t_join_ep_ins.create_day) as update_time ---修改为建组日期作为分区日期
        , t_join_ep_ins.sn_code
        , t_join_ep_ins.calculate_type
        , t_special_bill.next_scansite_code
        , t_special_bill.next_scansite_name
        , t_special_bill.shipment_no
        , t_special_bill.is_reback
        , t_special_bill.package_code
        , to_date(t_join_ep_ins.create_day) as dt
from t_special_bill
    left join
    (
    select *,
    row_number() over(partition by create_institution_id,institutiona_name order by sign_in_time) rk1, --给员工添加编码
    count(1)     over(partition by create_institution_id,institutiona_name) cnt1  --班组中有有拆包标识的员工个数
    from t_join_ep_ins
    where is_special = 1
    ) t_join_ep_ins --拆包员工的信息
on t_special_bill.create_institution_id = t_join_ep_ins.create_institution_id --班组对应
    and t_special_bill.rk%t_join_ep_ins.cnt1=t_join_ep_ins.rk1-1 --单号序号、员工个数取模=员工序号
    ),
    t_join_four_ep_ins as (
select * from t_other_scan
union all
select * from t_ass_scan
union ALL
select * from t_unpackage_scan_employ
union ALL
select * from t_special_bill_employ
    ),
    ---**计重挂靠计件
    --取出被挂靠的班组的单号
    be_rely_institution_bill as (
select bre.billcode
        , bre.scan_time
        , bre.scan_type
        , bre.center_code
        , bre.center_name
        , bre.institution_code
        , bre.institution_name
        , bre.next_scansite_code
        , bre.next_scansite_name
        , bre.shipment_no
        , bre.is_reback
        , bre.package_code
        , row_number() over(partition by bre.center_code,bre.institution_code order by bre.billcode) bill_rnk
from t_join_four_ep_ins bre
    inner join
    (  --取出被挂靠的计重班组
    select center_code,rely_institution_code
    from t_join_ep_ins
    where rely_institution_code is not null and calculate_type=2  --（1计件2计重）
    group by center_code,rely_institution_code
    )re on re.center_code=bre.center_code and re.rely_institution_code=bre.institution_code
    ),
    --挂靠的员工信息
    rely_emp_info as (
select  *
        , row_number() over (partition by center_code,institution_code order by employ_code) emp_rnk
        , count(1) over (partition by center_code,institution_code ) re_emp_cnt
from t_join_ep_ins
where rely_institution_code is not null and calculate_type=2  --（1计件2计重）
    ),
--将被挂靠组的单号赋值给挂靠组的员工
    rely_emp_bill as (
select  be.billcode
        , be.scan_time
    --, '计重挂靠计件' as scan_type --
        , be.scan_type
        , re.center_code
        , re.center_name
        , re.institution_code
        , re.institution_name
        , re.institutiona_code
        , re.institutiona_name
        , re.schedule_code
        , re.schedule_name
        , re.create_day
        , re.employ_code
        , re.employ_name
        , re.post_code
        , re.post_name
        , re.linked_method
        , re.sign_in_time
        , re.sign_out_time
        , re.work_hours
        , re.group_time_type
        , to_date(re.create_day) as update_time ---修改为建组日期作为分区日期
        , re.sn_code
        , re.calculate_type
        , be.next_scansite_code
        , be.next_scansite_name
        , be.shipment_no
        , be.is_reback
        , be.package_code
        , to_date(re.create_day) as dt
from be_rely_institution_bill be
    inner join rely_emp_info re
on be.center_code = re.center_code
    and be.institution_code = re.rely_institution_code
    and be.bill_rnk%re.re_emp_cnt=re.emp_rnk-1
    ),
    t_weight_billcode as (
select waybill_no
        , taking_max_weight
        , first_network_max_weight
        , first_center_max_weight
from (
    select waybill_no
        , taking_max_weight  --最大揽收重量
        , first_network_max_weight --始发网点最大重量
        , first_center_max_weight  --始发中心最大重量
        , dt
        ,row_number() over(partition by waybill_no order by dt) bill_rank
    from jms_dwd.dwd_wide_unsign_summary_waybill_dt
    where dt between date_add('{{ execution_date | date_add(-1) | cst_ds }}', -30) and date_add('{{ execution_date | date_add(-1) | cst_ds }}', 0)
    ) tmp1
where bill_rank = 1
    )

insert overwrite table  bidefault.dm_employee_operate_num_billcode_detail partition(dt)
select
    t1.billcode
     , t1.scan_time
     , t1.scan_type
     , t1.center_code
     , t1.center_name
     , t1.institution_code
     , t1.institution_name
     , t1.institutiona_code
     , t1.institutiona_name
     , t1.schedule_code
     , t1.schedule_name
     , t1.create_day
     , t1.employ_code
     , t1.employ_name
     , t1.post_code
     , t1.post_name
     , t1.linked_method
     , t1.sign_in_time
     , t1.sign_out_time
     , t1.work_hours
     , t1.group_time_type
     , t1.update_time ---修改为建组日期作为分区日期
     , t1.sn_code
     ,case when t1.calculate_type = 2 then coalesce( t2.first_center_max_weight,t2.taking_max_weight,t2.first_network_max_weight,0)
           else 0
    end weight
     , t1.calculate_type --结算类型添加
     , t1.next_scansite_code
     , t1.next_scansite_name
     , t1.shipment_no
     , t1.is_reback
     , if(split(t1.group_time_type,'#')[2] in ('转运中心装车','转运中心卸车') and package_code is not null,0,1) as is_pieces     --包+单件,件数
     , if(split(group_time_type,'#')[2] in ('转运中心装车','转运中心卸车') and substr(billcode,1,1)='B',0,1) as is_ticket   --包内件+单间，票数
     , t1.package_code
     , t1.dt
from (select * from t_join_four_ep_ins union all select * from rely_emp_bill) t1
         left join t_weight_billcode t2 on t1.billcode = t2.waybill_no
distribute by dt,abs(hash(group_time_type))%160;
------------------------------------------------------------------------------------------------------------------------
--员工操作量汇总+重量汇总+问题类型
with t_join_ep_ins as  --员工信息表
         (   select
                 t1.*,t2.sign_out_time_ye
             from
                 (
                     select *
                     from  bidefault.dm_employee_operate_info_detail
                     where dt ='{{ execution_date | date_add(-1) | cst_ds }}'
                 )t1
                     left join
                 ( --获取前天员工的最大签出时间
                     select center_code,employ_code,max(sign_out_time) sign_out_time_ye
                     from  bidefault.dm_employee_operate_info_detail
                     where dt = date_add('{{ execution_date | date_add(-1) | cst_ds }}',-1)
                     group by center_code,employ_code
                 ) t2 on t1.center_code=t2.center_code and t1.employ_code=t2.employ_code
         ),
     t_join_four_ep_ins as ( --员工单号信息
         select  *
         from bidefault.dm_employee_operate_num_billcode_detail
         where dt='{{ execution_date | date_add(-1) | cst_ds }}'
     ),
     t_all_oper as (--所有操作信息
         select * from bidefault.dm_employee_operate_num_all_detail where dt='{{ execution_date | date_add(-1) | cst_ds }}'
     ),

     t_group_count_total as ( --员工汇总量
         select center_code
              , max(center_name) as center_name
              , employ_code
              , max(employ_name) as employ_name
              , group_time_type
              , sum(is_pieces) as per_oper      --包+单件,件数
              , sum(is_ticket) as per_oper_vo  --包内件+单件，票数
              , sum(weight) as employ_weight --重量汇总
         from t_join_four_ep_ins
         group by center_code
                , employ_code
                , group_time_type
     )

insert overwrite table bidefault.dm_employee_operate_num_detail partition(dt)
select
    t2.center_code
     ,t2.center_name
     ,t2.institution_code
     ,t2.institution_name
     ,t2.institutiona_code
     ,t2.institutiona_name
     ,t2.schedule_code
     ,t2.schedule_name
     ,t2.employ_code
     ,t2.employ_name
     ,t2.post_code
     ,t2.post_name
     ,t2.linked_method
     ,t2.sign_in_time
     ,t2.sign_out_time
     ,t2.work_hours
     ,nvl(t1.per_oper,0) as per_oper
     ,to_date(t2.create_day) as update_time ---修改为建组日期作为分区日期
     ,t1.group_time_type                    --用于关联条件
     ,case when t1.employ_code is not null  then '有操作量的正常数据'
           when t3.scanuser_code is not null and t4.scan_type is null then '有操作记录，关联类型错误'
           when  t4.scan_type is not null then '有操作记录，签入签出时间不在时间范围内'
           else '无操作记录'
    end as question_type
     ,t2.create_day as create_day
     ,t2.sn_code
     ,t1.employ_weight --员工重量
     ,t2.calculate_type --结算类型
     ,t2.employ_type --员工类型
     ,t2.operate_code --操作类型1-操作/2-非操作
     ,t1.per_oper_vo --操作票数
     ,date(t2.create_day)  as dt
from  t_join_ep_ins t2
    left join t_group_count_total t1
on (t1.center_code = t2.center_code and t1.employ_code = t2.employ_code and t1.group_time_type = t2.group_time_type)
    left join
    (select scansite_code,scanuser_code,max(scan_time) scan_time_max from t_all_oper group by scansite_code,scanuser_code ) t3
    on t2.center_code = t3.scansite_code
    and t2.employ_code = t3.scanuser_code
    and t2.sign_out_time_ye<t3.scan_time_max
    left join
    (select scansite_code,scanuser_code,scan_type,max(scan_time) scan_time_max from t_all_oper group by scansite_code,scanuser_code,scan_type ) t4
    on t2.center_code = t4.scansite_code
    and t2.employ_code = t4.scanuser_code
    and t2.linked_method =t4.scan_type
    and t2.sign_out_time_ye<t4.scan_time_max
distribute by dt;